package cn.itcast.map;

import cn.itcast.bean.IndexBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import cn.itcast.util.DbUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.Date;
import java.util.Map;

public class IndexKlineMap extends RichMapFunction<IndexBean, Row> {
    private String kType;
    private String firstTxDateType;

    public IndexKlineMap(String type, String firstTxDateType) {
        this.kType=type;
        this.firstTxDateType=firstTxDateType;
    }

    String tradeDate = null;
    String firstTradeDate = null;
    Map<String, Map<String, Object>> klineMap=null;

    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         * 1.创建构造方法
         *   入参：kType：K线类型
         *         firstTxdate：周期首个交易日
         * 2.获取交易日历表交易日数据
         * 3.获取周期首个交易日
         * 4.获取不同K线下的表数据
         */

        //2.获取交易日历表交易日数据
        String sql = "select * from tcc_date where trade_date <= curdate() order by trade_date desc limit 1";
        Map<String, String> dateMap = DbUtil.queryKv(sql);
        tradeDate = dateMap.get("trade_date"); //最新交易日（T日）
        firstTradeDate = dateMap.get(firstTxDateType);//周期首个交易日

        String sqlKline = "select index_code,\n" +
                "       max(high_price) as high_price,\n" +
                "       min(low_price) as low_price,\n" +
                "       sum(trade_vol) as total_vol,\n" +
                "       sum(trade_amt) as total_amt\n" +
                "from bdp_quot_index_kline_day\n" +
                "where trade_date between "+firstTradeDate+" and "+tradeDate+" \n" +
                "group by 1;";    //1表示按照index_code分组

        //指数代码  key为字段名  value为字段值
        klineMap = DbUtil.query("index_code", sqlKline);

    }

    @Override
    public Row map(IndexBean indexBean) throws Exception {
        /**
         * 1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
         * 2.获取交易日时间
         * 3.比较周期首个交易日和当天交易日大小，判断是否是周、月K线
         * 4.获取周/月K数据：成交量、成交额、高、低
         * 5.高、低价格比较
         * 6.计算均价
         * 7.封装数据Row
         */

        //todo 1.获取个股部分数据（前收、收、开盘、高、低、量、金额）
        BigDecimal preClosePrice = indexBean.getPreClosePrice();
        BigDecimal closePrice = indexBean.getClosePrice();
        BigDecimal openPrice = indexBean.getOpenPrice();
        BigDecimal highPrice = indexBean.getHighPrice();
        BigDecimal lowPrice = indexBean.getLowPrice();
        Long tradeVolDay = indexBean.getTradeVolDay();
        Long tradeAmtDay = indexBean.getTradeAmtDay(); //总金额
        //todo 2.获取交易日时间,转换成long
        Long tradeTime = DateUtil.stringToLong(tradeDate, Constant.format_yyyy_mm_dd);
        Long firstTradeTime = DateUtil.stringToLong(firstTradeDate, Constant.format_yyyy_mm_dd);
        //todo 3.比较周期首个交易日和当天交易日大小，判断是否是周、月K线
        if(firstTradeTime<tradeTime && (kType.equals("2") || kType.equals("3"))){//是周K或月K
            //todo 4.获取周/月K数据：成交量、成交额、高、低
            Map<String, Object> map = klineMap.get(indexBean.getIndexCode());//历史数据
            if(map!=null && map.size()>0){
                //历史总成交量和总成交金额
                Long totalVol = Long.valueOf(map.get("total_vol").toString());
                Long totalAmt = Long.valueOf(map.get("total_amt").toString());
                //历史最高价和历史最低价
                BigDecimal high_price = new BigDecimal(map.get("high_price").toString());
                BigDecimal low_price = new BigDecimal(map.get("low_price").toString());
                tradeVolDay += totalVol; //总成交量=当前总成交量+历史总成交量
                tradeAmtDay += totalAmt;
                //todo 5.高、低价格比较
                if(high_price.compareTo(highPrice)==1){ //>
                    highPrice = high_price;
                }
                if(low_price.compareTo(lowPrice)==-1){ //<
                    lowPrice = low_price;
                }
            }
        }
        //todo 6.计算均价
        BigDecimal avgPrice = new BigDecimal(0);
        if(tradeVolDay!=0){
            avgPrice = new BigDecimal(tradeAmtDay).divide(new BigDecimal(tradeVolDay),2,RoundingMode.HALF_UP);
        }
        //todo 7.封装数据Row
        /**
         * update_time     timestamp       '更新时间',
         * trade_date      date             '交易日期',
         * index_code      char(6)          '指数代码',
         * index_name      varchar(20)      '指数名称',
         * k_line_type     char             'K线类型',
         * pre_close_price decimal(8, 2)    '前收盘价',
         * open_price      decimal(8, 2)    '开盘价',
         * high_price      decimal(8, 2)    '最高价',
         * low_price       decimal(8, 2)    '最低价',
         * close_price     decimal(8, 2)    '收盘价',
         * avgprice        decimal(8, 2)    '均价',
         * trade_vol       decimal(18)      '成交量',
         * trade_amt       decimal(18, 2)   '成交金额',
         */
        Row row = new Row(13);
        row.setField(0,new Timestamp(new Date().getTime()));
        row.setField(1,tradeDate);
        row.setField(2,indexBean.getIndexCode());
        row.setField(3,indexBean.getIndexName());
        row.setField(4,kType);
        row.setField(5,preClosePrice);
        row.setField(6,openPrice);
        row.setField(7,highPrice);
        row.setField(8,lowPrice);
        row.setField(9,closePrice);
        row.setField(10,avgPrice);
        row.setField(11,tradeVolDay);
        row.setField(12,tradeAmtDay);
        return row;
    }
}
